package replicated_calculator;
import MulticastQuene.*;
import point_to_point_queue.*;

import java.io.IOException;
import java.math.BigInteger;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.net.InetSocketAddress;

/**
 *
 * An stand-along implementation of a server. It simply keeps the variables in 
 * a hash map and executes the commands in the natural way on this hash map.
 *
 * @author Jesper Buus Nielsen, Aarhus University, 2012.
 *
 */

public class ServerReplicated extends Thread implements ClientEventVisitor, Server {

    /** Map containing all currently assigned variables */
    protected final HashMap<String,BigInteger> valuation= new HashMap<String,BigInteger>();
    /** A map of login names and their address */
    protected final HashMap<String,PointToPointQueueSenderEndNonRobust<ClientEvent>> clients;
    /** Pending operations from clients */
    private PointToPointQueueReceiverEndNonRobust<ClientEvent> operationsFromClients;
    /** The MulticastQueue for sending operations between servers */
    private MulticastQueue<ClientEvent> serverOperations;
    /** A list containing all events */
    ArrayList<ClientEvent> history;
    /** the host address of this server */
    String localHostAddress;


    int listenPortForClients = Parameters.mainServerPortForClients; //if left to default, this is the first server
    int listenPortForServers = Parameters.mainServerPortForServers;

    private MulticastQueue.DeliveryGuarantee deliveryGuarantee;

    /** If log messages should be printed or not */
    public boolean debug = Parameters.debugServer;

    /**
     * Constructor which handles all initialization
     */
    public ServerReplicated(){
        try {
            localHostAddress = InetAddress.getLocalHost().getCanonicalHostName();
        }
        catch (UnknownHostException e) {
            System.out.println("Could not get local address");
            e.printStackTrace();
            System.exit(-1);
        }

        history = new ArrayList<ClientEvent>();
        clients = new HashMap<String,PointToPointQueueSenderEndNonRobust<ClientEvent>>();
        serverOperations = new MulticastQueueTotalOnly_v2<ClientEvent>();
        deliveryGuarantee = MulticastQueue.DeliveryGuarantee.TOTAL;
    }


//	********************************************************************
//	**********************   INTERFACE METHODS   ***********************
//	********************************************************************

    public void createGroup(int port) {
        try {
            operationsFromClients = new PointToPointQueueReceiverEndNonRobust<ClientEvent>();
            operationsFromClients.listenOnPort(listenPortForClients);

            //Opens port Parameters.mainServerPortForServers
            serverOperations.createGroup( listenPortForServers, deliveryGuarantee);
        } catch (IOException e) {
            System.err.println("Cannot start server!");
            System.err.println("Check your network connection!");
            System.err.println("Check that no other service runs on port " + listenPortForClients + "!");
            System.err.println();
            System.err.println(e);
            System.exit(-1);
        }
        this.start();
    }


    public void joinGroup(InetSocketAddress knownPeer) {
        //Open a channel for recieving the history from one of the other servers.
        PointToPointQueueReceiverEnd<ClientEvent> p2pHistoryReciever =  new PointToPointQueueReceiverEndNonRobust<ClientEvent>();
        try{
            operationsFromClients = new PointToPointQueueReceiverEndNonRobust<ClientEvent>();
            operationsFromClients.listenOnPort(listenPortForClients);

            p2pHistoryReciever.listenOnPort(Parameters.historyPort);
        } catch (IOException e) {
            System.err.println("Error opening the port used for transferring history: " + Parameters.historyPort);
            e.printStackTrace();
            System.exit(-1);
        }

        log("Joining Server network");
        //open connection to the other server
        try {
            serverOperations.joinGroup(listenPortForServers, knownPeer, deliveryGuarantee);
        } catch (IOException e) {
            System.err.println("Cannot start server!");
            System.err.println("Check your network connection!");
            System.err.println("Check that no other service runs on port " + listenPortForClients + "!");
            System.err.println(e);
            System.exit(-1);
        }


        //Get the older messages, and visit them
        log("Waiting for history transmission on " + localHostAddress + ":" + Parameters.historyPort);
        ClientEvent msg;
        while((msg = p2pHistoryReciever.get()) != null){
            log("HISTORY element gotten");
            if(msg instanceof ClientEventEoH){
                //an message of this type signals "End of History"
                log("Transmission done, recieved: " + history.size());
                break;
            }

            history.add(msg);
            msg.accept(this);
        }
        log("HISTORY GOTTEN");

        p2pHistoryReciever.shutdown();


        //Start the server
        this.start();
    }


    public void leaveGroup() {
        operationsFromClients.shutdown();
        for (String client : clients.keySet()) {
            clients.get(client).shutdown();
        }
        clients.clear();

        serverOperations.leaveGroup();
    }

    @Override
    public void setListenPort(int listenPortForServers, int listenPortForClients) {
        this.listenPortForServers = listenPortForServers;
        this.listenPortForClients = listenPortForClients;
        log("Ports changed. Server listen port: " + listenPortForServers + " client listen port: " + listenPortForClients);
    }

//	********************************************************************
//	************************    RUN METHOD    **************************
//	********************************************************************

    /**
     * Keeps getting the next operation from a server and then visits the operation.
     * Sub-classes have to implement the visiting methods.
     */
    public void run() {
        log("Main thread started");
        Thread clientThread = new Thread(new ClientThread(this));
        Thread operationThread = new Thread(new OperationThread(this));
        clientThread.start();
        operationThread.start();
        try {
            clientThread.join();
            operationThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        log("Main thread stopped");
    }



//	********************************************************************
//	**********************    PRIVATE THREADS    ***********************
//	********************************************************************

    /**
     * Thread handling the incoming operations from all clients
     */
    private class ClientThread implements Runnable{
        private ServerReplicated server;

        public ClientThread(ServerReplicated serverReplicated){
            server = serverReplicated;
        }

        @Override
        public void run() {
            log("Started");
            ClientEvent nextOperation;

            //Fetch operations until queue returns null indicating a shutdown
            while ((nextOperation = operationsFromClients.get())!=null) {

                if(nextOperation instanceof ClientEventConnect){
                    nextOperation.accept(server);
                }
                else{
                    serverOperations.put(nextOperation);
                }



            }
            log("Stopped");

        }

        private void log(String msg){
            server.log("Client Thread: " + msg);
        }
    }

    /**
     * Thread which handles all new operations
     */
    private class OperationThread implements Runnable{
        ServerReplicated server;

        public OperationThread(ServerReplicated serverReplicated){
            server = serverReplicated;
        }

        @Override
        public void run() {
            log("Started");

            //keep polling for messages until null is recieved (null signals shutdown)
            MulticastMessage msg;
            while((msg = serverOperations.get()) != null){
                try {
                    sleep(300);
                } catch (InterruptedException e) {
                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
                }

                log("Server Recieved: " + msg);

                if(msg instanceof MultiCastMessagePayloadTotal){ //Payload message
                    ClientEvent nextOperation = ((MultiCastMessagePayloadTotal<ClientEvent>) msg).getPayload();

                    if(nextOperation instanceof ClientEventRead){
                        log("Message was a PayloadRead");
                        boolean ourClient;
                        synchronized(clients) {
                            ourClient = clients.containsKey(nextOperation.clientName);
                        }
                        if(ourClient){
                            nextOperation.accept(server);
                        }
                    }
                    else if(nextOperation instanceof ClientEventDisconnect){
                        synchronized(clients) {
                            PointToPointQueueSenderEndNonRobust<ClientEvent> toClient = clients.get(nextOperation.clientName);
                            if (toClient != null) { //It will be null for servers that are not the client's home server
                                toClient.put(nextOperation);
                                nextOperation.accept(server);
                            }
                        }
                    }
                    else if(nextOperation instanceof ClientEventDisconnect || nextOperation instanceof ClientEventConnect) {
                        System.out.println("EVIL " + nextOperation);
                        System.exit(-1);
                    }
                    else{
                        log("Message was a Payload (not)Read");
                        //add to history
                        history.add(nextOperation);
                        log("Current log length: " + history.size());

                        //visit the operation
                        log("Visiting the operation: " + nextOperation);
                        nextOperation.accept(server);
                    }
                }
                else if(msg instanceof MultiCastMessageJoinTotal){ //A new server has joined, send history
                    MultiCastMessageJoinTotal jMsg = (MultiCastMessageJoinTotal) msg;
                    InetAddress joinerAddress = jMsg.getSender().getAddress();

                    //if its the servers own join message, do not do anything.
                    InetSocketAddress thisServerAddress = new InetSocketAddress(localHostAddress, listenPortForServers);
                    log("Address of this server: " + thisServerAddress);
                    log("Address of joiner: " + jMsg.getSender());
                    if( !thisServerAddress.equals(jMsg.getSender()) ){

                        PointToPointQueueSenderEnd<ClientEvent> outgoingHistoryQueue =
                                new PointToPointQueueSenderEndNonRobust<ClientEvent>();
                        InetSocketAddress joinerHistoryAddress =
                                new InetSocketAddress(joinerAddress, Parameters.historyPort);
                        outgoingHistoryQueue.setReceiver(joinerHistoryAddress);
                        log("Sending history to: " + joinerHistoryAddress);

                        for(ClientEvent oldEvent : history){
                            outgoingHistoryQueue.put(oldEvent);
                        }

                        //send a "End of History" special ClientEvent and shutdown the queue
                        outgoingHistoryQueue.put(new ClientEventEoH(localHostAddress));
                        outgoingHistoryQueue.shutdown();
                        log("Sent " + history.size());
                    }
                }
                else{
                    //System.out.println("WEIRD MESSAGE");
                }
            }
            log("Stopped");
        }


        private void log(String msg){
            server.log("Operation Thread: " + msg);
        }
    }


//	********************************************************************
//	**********************    PRIVATE METHODS    ***********************
//	********************************************************************

    @Override
    public String toString() {
        return "ServerReplicated{" +
                "listenPortForServers=" + listenPortForServers +
                ", listenPortForClients=" + listenPortForClients +
                ", localHostAddress='" + localHostAddress + '\'' +
                ", historySize=" + history.size() +
                ", debug=" + debug +
                '}';
    }


    private void log(String message){
        Calendar instance = Calendar.getInstance();
        String time = instance.get(Calendar.HOUR)
                + ":" + instance.get(Calendar.MINUTE)
                + ":" + instance.get(Calendar.SECOND);
        if(debug) System.out.println(time + " " + this.getName() + " " + message);
    }


    protected BigInteger valuate(String var) {
        BigInteger val;
        synchronized(valuation) {
            val = valuation.get(var);
        }
        if (val==null) {
            val = BigInteger.ZERO;
        }
        return val;
    }


    /**
     * Print output to client
     * @param event Originating event
     */
    protected void acknowledgeEvent(ClientEvent event) {
        synchronized(clients) {
            PointToPointQueueSenderEndNonRobust<ClientEvent> toClient = clients.get(event.clientName);
            if (toClient != null) { //It will be null for servers that are not the client's home server
                toClient.put(event);
            }
        }
    }

//	********************************************************************
//	**********************    VISITOR METHODS    ***********************
//	********************************************************************

    /**
     * Connects a client given a connection event from the client.
     * This is done
     *
     * @param eventConnect
     */
    public void visit(ClientEventConnect eventConnect) {
        final String clientName = eventConnect.clientName;
        final InetSocketAddress clientAddress = eventConnect.clientAddress;
        PointToPointQueueSenderEndNonRobust<ClientEvent> queueToClient = new PointToPointQueueSenderEndNonRobust<ClientEvent>();
        queueToClient.setReceiver(clientAddress);
        synchronized(clients) {
            clients.put(clientName,queueToClient);
            acknowledgeEvent(eventConnect);
        }
    }


    /**
     * Disconnects a client on a disconnect event from the client.
     *
     * @param eventDisconnect
     */
    public void visit(ClientEventDisconnect eventDisconnect) {
        synchronized(clients) {
            PointToPointQueueSenderEndNonRobust<ClientEvent> queueToClient = clients.remove(eventDisconnect.clientName);
            queueToClient.shutdown();
        }
    }


    public void visit(ClientEventAdd eventAdd) {
        synchronized(valuation) {
            valuation.put(eventAdd.res, valuate(eventAdd.left).add(valuate(eventAdd.right)));
            acknowledgeEvent(eventAdd);
        }
    }


    public void visit(ClientEventAssign eventAssign) {
        synchronized(valuation) {
            valuation.put(eventAssign.var, eventAssign.val);
            acknowledgeEvent(eventAssign);
        }
    }


    public void visit(ClientEventCompare eventCompare) {
        synchronized(valuation) {
            valuation.put(eventCompare.res, BigInteger.valueOf(valuate(eventCompare.left).compareTo(valuate(eventCompare.right))));
            acknowledgeEvent(eventCompare);
        }
    }


    /**
     * Multiplies to variables on a multiplication event from the client.
     *
     * @param eventMult The multiplication event from the client.
     */
    public void visit(ClientEventMult eventMult) {
        synchronized(valuation) {
            valuation.put(eventMult.res, valuate(eventMult.left).multiply(valuate(eventMult.right)));
            acknowledgeEvent(eventMult);
        }
    }


    /**
     * Reads a variable on a read event from the client.
     *
     * @param eventRead The read event from the client.
     */
    public void visit(ClientEventRead eventRead) {
        synchronized(valuation) {
            eventRead.setVal(valuate(eventRead.var));
            log("Reading var:" + eventRead.var + ":" + eventRead.getVal());
            acknowledgeEvent(eventRead);
        }
    }



//	********************************************************************
//	**********************      UNSUPPORTED      ***********************
//	********************************************************************

    /**
     * Not supported yet.
     * @param _
     */
    public void visit(ClientEventBeginAtomic _) {
        throw new UnsupportedOperationException("ClientEventBeginAtomic");
        //Not needed
    }


    /**
     * Not supported yet.
     * @param _
     */
    public void visit(ClientEventEndAtomic _) {
        throw new UnsupportedOperationException("ClientEventEndAtomic");
    }
}
